
#include "cgzClickHouse.h"
#include "cgzCThreadPool.h"

cgzClickHouseReply::cgzClickHouseReply(cgzClickHouseCode code)
    : m_code(code)
{
}

cgzClickHouseReply::~cgzClickHouseReply()
{
}

// 执行命令基类
cgzClickHouseExec::cgzClickHouseExec(const cgzClickHouseCmd cmd, const int cookid)
    : cgzClickHouseReply(cgzClickHouseCode::EXEC_REQUEST),
      m_cmd(cmd),
      m_cookid(cookid),
      m_ret(false)
{
}

cgzClickHouseExec::~cgzClickHouseExec()
{
}

// 插入数据指令
cgzClickHouseInsertBlock::cgzClickHouseInsertBlock(
    const char *tbname,
    std::unique_ptr<clickhouse::Block> &&block,
    const long cookid)
    : cgzClickHouseExec(cgzClickHouseCmd::INSERT_BLOCK, cookid),
      m_tbname(tbname),
      m_block(std::move(block))
{
}

cgzClickHouseInsertBlock::~cgzClickHouseInsertBlock()
{
}

// 语句插入指令
cgzClickHouseInsert::cgzClickHouseInsert(
    const char *sql,
    const int cookid)
    : cgzClickHouseExec(cgzClickHouseCmd::INSERT, cookid),
      m_sql(sql)
{
}

cgzClickHouseInsert::~cgzClickHouseInsert()
{
}

// 更新数据指令
cgzClickHouseUpdate::cgzClickHouseUpdate(
    const char *sql,
    const int cookid)
    : cgzClickHouseExec(cgzClickHouseCmd::UPDATE, cookid),
      m_sql(sql)
{
}

cgzClickHouseUpdate::~cgzClickHouseUpdate()
{
}

// 删除数据指令
cgzClickHouseDelete::cgzClickHouseDelete(
    const char *sql,
    const int cookid)
    : cgzClickHouseExec(cgzClickHouseCmd::DELETE, cookid),
      m_sql(sql)
{
}

cgzClickHouseDelete::~cgzClickHouseDelete()
{
}

// 查询数据指令
cgzClickHouseSelect::cgzClickHouseSelect(
    const char *sql,
    const int cookid)
    : cgzClickHouseExec(cgzClickHouseCmd::SELECT, cookid),
      m_sql(sql),
      m_block(nullptr)
{
}

cgzClickHouseSelect::~cgzClickHouseSelect()
{
}

// 创建指令
cgzClickHouseCreate::cgzClickHouseCreate(
    const char *sql,
    const int cookid)
    : cgzClickHouseExec(cgzClickHouseCmd::CREATE, cookid),
      m_sql(sql)
{
}

cgzClickHouseCreate::~cgzClickHouseCreate()
{
}

// drop指令
cgzClickHouseDrop::cgzClickHouseDrop(
    const char *sql,
    const int cookid)
    : cgzClickHouseExec(cgzClickHouseCmd::DROP, cookid),
      m_sql(sql)
{
}

cgzClickHouseDrop::~cgzClickHouseDrop()
{
}

// 修改指令
cgzClickHouseAlter::cgzClickHouseAlter(
    const char *sql,
    const int cookid)
    : cgzClickHouseExec(cgzClickHouseCmd::ALTER, cookid),
      m_sql(sql)
{
}

cgzClickHouseAlter::~cgzClickHouseAlter()
{
}

// 截断指令
cgzClickHouseTruncate::cgzClickHouseTruncate(
    const char *sql,
    const int cookid)
    : cgzClickHouseExec(cgzClickHouseCmd::TRUNCATE, cookid),
      m_sql(sql)
{
}

cgzClickHouseTruncate::~cgzClickHouseTruncate()
{
}

// 执行计划
cgzClickHouseExplain::cgzClickHouseExplain(
    const char *sql,
    const int cookid)
    : cgzClickHouseSelect(sql, cookid)
{
    m_cmd = cgzClickHouseCmd::EXPLAIN;
}

cgzClickHouseExplain::~cgzClickHouseExplain()
{
}

// 显示查询
cgzClickHouseShow::cgzClickHouseShow(
    const char *sql,
    const int cookid)
    : cgzClickHouseSelect(sql, cookid)
{
    m_cmd = cgzClickHouseCmd::SHOW;
}

cgzClickHouseShow::~cgzClickHouseShow()
{
}

// 切换指令
cgzClickHouseUse::cgzClickHouseUse(
    const char *sql,
    const int cookid)
    : cgzClickHouseExec(cgzClickHouseCmd::USE, cookid),
      m_sql(sql)
{
}

cgzClickHouseUse::~cgzClickHouseUse()
{
}

cgzClickHouse::cgzClickHouse(cgzCThreadPool *pool,
                             const char *name,
                             const char *ip,
                             const char *user,
                             const char *password,
                             const int port,
                             const char *db,
                             const int cid)
    : cgzITask(pool, name, cgzETask::ClickHouse),
      m_ip(ip),
      m_user(user),
      m_password(password),
      m_firstAuth(false),
      m_ping(false),
      m_db(db),
      m_port(port),
      m_cid(cid)
{
    connect();
    service();
}

cgzClickHouse::~cgzClickHouse()
{
    if (m_client)
    {
        m_client->ResetConnection();
    }
}

void cgzClickHouse::authret()
{
    m_ping = true;
    if (!m_firstAuth)
    {
        m_firstAuth = true;
        auto node = malloc(sizeof(cgzClickHouseReply));
        auto acptC = new (node) cgzClickHouseReply(cgzClickHouseCode::AUTH_SUCCESS);
        acptC->target = m_cid;
        acptC->origin = getId();
        retTask(acptC);
    }
}

bool cgzClickHouse::connect()
{
    try
    {
        clickhouse::ClientOptions opts;
        opts.SetHost(m_ip);
        opts.SetPort(m_port);
        opts.SetUser(m_user);
        opts.SetPassword(m_password);
        opts.SetDefaultDatabase(m_db);
        opts.SetCompressionMethod(clickhouse::CompressionMethod::LZ4);
        opts.SetSendRetries(3);
        opts.SetPingBeforeQuery(true);

        m_client = std::make_unique<clickhouse::Client>(opts);
        m_client->Ping(); // 测试连接
        authret();
        cgz_this_error("Connected to ClickHouse {} at {}:{}", m_name, m_ip, m_port);
        return true;
    }
    catch (const clickhouse::ServerException &e)
    {
        cgz_this_error("Connected failed ClickHouse {} at {}:{}", m_name, m_ip, m_port);
        return false;
    }
    catch (const std::exception &e)
    {
        cgz_this_error("Connected failed ClickHouse {} at {}:{}", m_name, m_ip, m_port);
        return false;
    }
}

bool cgzClickHouse::rconnect()
{
    try
    {
        m_client->Ping(); // 测试连接
        return true;
    }
    catch (const clickhouse::ServerException &e)
    {
        cgz_this_error("Ping failed: {}", e.what());
        return connect();
    }
    catch (const std::exception &e)
    {
        return connect();
    }
}

/// @brief 合并深拷贝
/// @param bFirst 是否第一次回调
/// @param src    当前回调数据
/// @param dst    合并目标数据
void deepBlock(bool bFirst, const clickhouse::Block &src, clickhouse::Block &dst)
{
    int colCount = src.GetColumnCount();
    if (bFirst)
    {
        // 首次填充列名字
        for (int i = 0; i < colCount; ++i)
        {
            auto col = src[i];
            auto cmd = col->Type()->GetCode();
            auto &colName = src.GetColumnName(i);
            switch (cmd)
            {
            case clickhouse::Type::Int8:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnInt8>());
            }
            break;
            case clickhouse::Type::Int16:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnInt16>());
            }
            break;
            case clickhouse::Type::Int32:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnInt32>());
            }
            break;
            case clickhouse::Type::Int64:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnInt64>());
            }
            break;
            case clickhouse::Type::UInt8:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnUInt8>());
            }
            break;
            case clickhouse::Type::UInt16:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnUInt16>());
            }
            break;
            case clickhouse::Type::UInt32:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnUInt32>());
            }
            break;
            case clickhouse::Type::UInt64:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnUInt64>());
            }
            break;
            case clickhouse::Type::Float32:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnFloat32>());
            }
            break;
            case clickhouse::Type::Float64:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnFloat64>());
            }
            break;
            case clickhouse::Type::String:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnString>());
            }
            break;
            case clickhouse::Type::FixedString:
            {
                auto size = col->As<clickhouse::ColumnFixedString>()->FixedSize();
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnFixedString>(size));
            }
            break;
            case clickhouse::Type::Date:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnDate>());
            }
            break;
            case clickhouse::Type::DateTime:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnDateTime>());
            }
            break;
            case clickhouse::Type::UUID:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnUUID>());
            }
            break;
            case clickhouse::Type::IPv4:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnIPv4>());
            }
            break;
            case clickhouse::Type::IPv6:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnIPv6>());
            }
            break;
            case clickhouse::Type::Int128:
            {
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnInt128>());
            }
            break;
            case clickhouse::Type::Decimal:
            {
                auto precision = col->As<clickhouse::ColumnDecimal>()->GetPrecision();
                auto scale = col->As<clickhouse::ColumnDecimal>()->GetScale();
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnDecimal>(precision, scale));
            }
            break;
            case clickhouse::Type::LowCardinality:
            {
                auto ref = col->As<clickhouse::ColumnLowCardinality>()->CloneEmpty();
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnLowCardinality>(ref));
            }
            break;
            case clickhouse::Type::DateTime64:
            {
                auto arg = col->As<clickhouse::ColumnDateTime64>()->GetPrecision();
                dst.AppendColumn(colName, std::make_shared<clickhouse::ColumnDateTime64>(arg));
            }
            break;
            case clickhouse::Type::Enum8:
            {
                // 获取枚举列并验证类型
                auto enum_col = col->As<clickhouse::ColumnEnum8>();
                // 获取枚举类型定义
                auto enum_type = enum_col->Type()->As<clickhouse::EnumType>();
                // 构建新的枚举项列表
                std::vector<clickhouse::Type::EnumItem> new_enum_items;
                for (auto it = enum_type->BeginValueToName(); it != enum_type->EndValueToName(); ++it)
                {
                    new_enum_items.push_back({
                        it->second.data(),              // name
                        static_cast<int16_t>(it->first) // value
                    });
                }

                // 创建新列
                auto new_col = std::make_shared<clickhouse::ColumnEnum8>(
                    clickhouse::Type::CreateEnum8(new_enum_items));

                // 添加到目标Block
                dst.AppendColumn(colName, new_col);
            }
            break;
            case clickhouse::Type::Enum16:
            {
                // 获取枚举列并验证类型
                auto enum_col = col->As<clickhouse::ColumnEnum16>();
                // 获取枚举类型定义
                auto enum_type = enum_col->Type()->As<clickhouse::EnumType>();
                // 构建新的枚举项列表
                std::vector<clickhouse::Type::EnumItem> new_enum_items;
                for (auto it = enum_type->BeginValueToName(); it != enum_type->EndValueToName(); ++it)
                {
                    new_enum_items.push_back({
                        it->second.data(),              // name
                        static_cast<int16_t>(it->first) // value
                    });
                }

                // 创建新列（带数据拷贝）
                auto new_col = std::make_shared<clickhouse::ColumnEnum16>(
                    clickhouse::Type::CreateEnum8(new_enum_items));

                // 添加到目标Block
                dst.AppendColumn(colName, new_col);
            }
            break;
            default:
            {

                assert(false);
            }
            }
        }
    }

    // 填充数据
    int rowCount = src.GetRowCount();
    for (int i = 0; i < rowCount; ++i)
    {
        for (int j = 0; j < colCount; ++j)
        {
            const auto &colSrc = src[j];
            const auto &colDst = dst[j];
            auto cmd = colSrc->Type()->GetCode();
            switch (cmd)
            {
            case clickhouse::Type::Int8:
            {
                auto v = colSrc->As<clickhouse::ColumnInt8>()->At(i);
                colDst->As<clickhouse::ColumnInt8>()->Append(v);
            }
            break;
            case clickhouse::Type::Int16:
            {
                auto v = colSrc->As<clickhouse::ColumnInt16>()->At(i);
                colDst->As<clickhouse::ColumnInt16>()->Append(v);
            }
            break;
            case clickhouse::Type::Int32:
            {
                auto v = colSrc->As<clickhouse::ColumnInt32>()->At(i);
                colDst->As<clickhouse::ColumnInt32>()->Append(v);
            }
            break;
            case clickhouse::Type::Int64:
            {
                auto v = colSrc->As<clickhouse::ColumnInt64>()->At(i);
                colDst->As<clickhouse::ColumnInt64>()->Append(v);
            }
            break;
            case clickhouse::Type::UInt8:
            {
                auto v = colSrc->As<clickhouse::ColumnUInt8>()->At(i);
                colDst->As<clickhouse::ColumnUInt8>()->Append(v);
            }

            break;
            case clickhouse::Type::UInt16:
            {
                auto v = colSrc->As<clickhouse::ColumnUInt16>()->At(i);
                colDst->As<clickhouse::ColumnUInt16>()->Append(v);
            }
            break;
            case clickhouse::Type::UInt32:
            {
                auto v = colSrc->As<clickhouse::ColumnUInt32>()->At(i);
                colDst->As<clickhouse::ColumnUInt32>()->Append(v);
            }
            break;
            case clickhouse::Type::UInt64:
            {
                auto v = colSrc->As<clickhouse::ColumnUInt64>()->At(i);
                colDst->As<clickhouse::ColumnUInt64>()->Append(v);
            }
            break;
            case clickhouse::Type::Float32:
            {
                auto v = colSrc->As<clickhouse::ColumnFloat32>()->At(i);
                colDst->As<clickhouse::ColumnFloat32>()->Append(v);
            }
            break;
            case clickhouse::Type::Float64:
            {
                auto v = colSrc->As<clickhouse::ColumnFloat64>()->At(i);
                colDst->As<clickhouse::ColumnFloat64>()->Append(v);
            }
            break;
            case clickhouse::Type::String:
            {
                auto v = colSrc->As<clickhouse::ColumnString>()->At(i);
                std::string s(v.data(), v.size());
                colDst->As<clickhouse::ColumnString>()->Append(std::move(s));
            }
            break;
            case clickhouse::Type::FixedString:
            {
                auto v = colSrc->As<clickhouse::ColumnFixedString>()->At(i);
                std::string s(v.data(), v.size());
                colDst->As<clickhouse::ColumnFixedString>()->Append(std::move(s));
            }
            break;
            case clickhouse::Type::Date:
            {
                auto v = colSrc->As<clickhouse::ColumnDate>()->At(i);
                colDst->As<clickhouse::ColumnDate>()->Append(v);
            }
            break;
            case clickhouse::Type::DateTime:
            {
                auto v = colSrc->As<clickhouse::ColumnDateTime>()->At(i);
                colDst->As<clickhouse::ColumnDateTime>()->Append(v);
            }
            break;
            case clickhouse::Type::UUID:
            {
                auto v = colSrc->As<clickhouse::ColumnUUID>()->At(i);
                colDst->As<clickhouse::ColumnUUID>()->Append(v);
            }
            break;
            case clickhouse::Type::IPv4:
            {
                auto v = colSrc->As<clickhouse::ColumnIPv4>()->At(i);
                colDst->As<clickhouse::ColumnIPv4>()->Append(v);
            }
            break;
            case clickhouse::Type::IPv6:
            {
                auto v = colSrc->As<clickhouse::ColumnIPv6>()->At(i);
                colDst->As<clickhouse::ColumnIPv6>()->Append(v);
            }
            break;
            case clickhouse::Type::Int128:
            {
                auto v = colSrc->As<clickhouse::ColumnInt128>()->At(i);
                colDst->As<clickhouse::ColumnInt128>()->Append(v);
            }
            break;
            case clickhouse::Type::Decimal:
            {
                auto v = colSrc->As<clickhouse::ColumnDecimal>()->At(i);
                colDst->As<clickhouse::ColumnDecimal>()->Append(v);
            }
            break;
            case clickhouse::Type::DateTime64:
            {
                auto v = colSrc->As<clickhouse::ColumnDateTime64>()->At(i);
                colDst->As<clickhouse::ColumnDateTime64>()->Append(v);
            }
            break;
            case clickhouse::Type::Enum8:
            {
                auto v = colSrc->As<clickhouse::ColumnEnum8>()->At(i);
                colDst->As<clickhouse::ColumnEnum8>()->Append(v);
            }
            break;
            case clickhouse::Type::Enum16:
            {
                auto v = colSrc->As<clickhouse::ColumnEnum16>()->At(i);
                colDst->As<clickhouse::ColumnEnum16>()->Append(v);
            }
            break;
            default:
            {

                assert(false);
            }
            }
        }
    }
}

bool cgzClickHouse::run(int64_t nms)
{
    bool no = false;
    while (!m_runTask.empty())
    {
        bool ok = true;
        auto task = static_cast<cgzClickHouseExec *>(m_runTask.front());

        if (!m_ping)
        {
            if (rconnect())
            {
                continue; // 重连成功
            }
            else
            {
                no = true; // 重连失败
            }
        }
        else if (no)
        {
            // 无法重连成
            ok = false;
        }
        else
        {
            try
            {
                switch (task->m_cmd)
                {
                // 插入数据
                case cgzClickHouseCmd::INSERT_BLOCK:
                {
                    auto pTask = static_cast<cgzClickHouseInsertBlock *>(task);
                    m_client->Insert(pTask->m_tbname, *(pTask->m_block));
                }
                break;
                // 插入数据
                case cgzClickHouseCmd::INSERT:
                {
                    auto pTask = static_cast<cgzClickHouseInsert *>(task);
                    m_client->Execute(pTask->m_sql);
                }
                break;
                // 更新数据
                case cgzClickHouseCmd::UPDATE:
                {
                    auto pTask = static_cast<cgzClickHouseUpdate *>(task);
                    m_client->Execute(pTask->m_sql);
                }
                break;
                // 删除数据
                case cgzClickHouseCmd::DELETE:
                {
                    auto pTask = static_cast<cgzClickHouseDelete *>(task);
                    m_client->Execute(pTask->m_sql);
                }
                break;
                // 创建结构
                case cgzClickHouseCmd::CREATE:
                {
                    auto pTask = static_cast<cgzClickHouseDelete *>(task);
                    m_client->Execute(pTask->m_sql);
                }
                break;
                // 丢弃结构
                case cgzClickHouseCmd::DROP:
                {
                    auto pTask = static_cast<cgzClickHouseDrop *>(task);
                    m_client->Execute(pTask->m_sql);
                }
                break;
                // 修改结构
                case cgzClickHouseCmd::ALTER:
                {
                    auto pTask = static_cast<cgzClickHouseAlter *>(task);
                    m_client->Execute(pTask->m_sql);
                }
                break;
                // 截断数据
                case cgzClickHouseCmd::TRUNCATE:
                {
                    auto pTask = static_cast<cgzClickHouseTruncate *>(task);
                    m_client->Execute(pTask->m_sql);
                }
                break;
                // 切换数据
                case cgzClickHouseCmd::USE:
                {
                    auto pTask = static_cast<cgzClickHouseUse *>(task);
                    m_client->Execute(pTask->m_sql);
                }
                break;
                // 执行计划
                case cgzClickHouseCmd::EXPLAIN:
                // 执行计划
                case cgzClickHouseCmd::SHOW:
                // 查询数据
                case cgzClickHouseCmd::SELECT:
                {
                    auto pTask = static_cast<cgzClickHouseSelect *>(task);
                    m_client->Select(pTask->m_sql,
                                     [pTask](const clickhouse::Block &block)
                                     {
                                         bool first = false;
                                         if (!pTask->m_block)
                                         {
                                             first = true;
                                             pTask->m_block = std::make_unique<clickhouse::Block>();
                                         }
                                         deepBlock(first, block, *(pTask->m_block));
                                     });
                }
                break;
                default:
                {
                    ok = false;
                    cgz_this_error("unknown cmd:{}", (int)task->m_cmd);
                }
                }
            }
            catch (const clickhouse::ServerException &e)
            {
                ok = false;
                cgz_this_error("Query failed (server): {} | SQL: {}", e.what(), (int)task->m_cmd);
            }
            catch (const std::exception &e)
            {
                ok = false;
                cgz_this_error("Query failed: {}", e.what());
            }
        }

        if ((!ok) && (!no))
        {
            // 执行错误-尝试一次重新连接
            m_ping = false;
            continue;
        }

        if (task->m_ret)
        {
            // 设置状态并回调
            task->m_err = !ok;
            task->origin = m_id;
            task->target = m_cid;
            task->m_code = cgzClickHouseCode::EXEC_REPLY;
            retTask(task);
        }
        else
        {
            task->~cgzClickHouseExec();
            free(task);
        }
        m_runTask.pop_front();
    }
    return true;
}

bool cgzClickHouse::addTask(cgzIData *task)
{
    return cgzITask::addTask(task);
}
